package com.open.fei.controller;

import java.util.ArrayList;
import java.util.List;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 生产者产生消息
 * @author zhaoYuFei
 */
@RestController
public class RocketMQController {


    @Autowired
    private RocketMQTemplate rocketMQTemplate;


    /**
     * 同步发送消息
     */
    @GetMapping("/provide/sends")
    public void sends(){
        SendResult sendResult = rocketMQTemplate.syncSend("powernode", "我是一个简单的消息");
        // 拿到消息的发送状态
        System.out.println(sendResult.getSendStatus());
        // 拿到消息的id
        System.out.println(sendResult.getMsgId());
    }
    /**
     * 异步发送消息
     */
    @GetMapping("/provide/sendAsync")
    public void sendAsync(){
        rocketMQTemplate.asyncSend("powernode","异步消息！",new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败");
            }
        });

        System.out.println("成功！");
    }
    /**
     * 单项发送消息
     */
    @GetMapping("/provide/sendOneWay")
    public void sendOneWay(){
        rocketMQTemplate.sendOneWay("powernode","单发");
    }
    /**
     * 延时发送消息
     */
    @GetMapping("/provide/sendDelay")
    public void sendDelay(){
        // 构建消息对象
        Message<String> message = MessageBuilder.withPayload("我是一个延迟消息").build();
        // 发送一个延时消息，延迟等级为4级，也就是30s后被监听消费
        SendResult sendResult = rocketMQTemplate.syncSend("powernode", message, 2000, 4);
        System.out.println(sendResult.getSendStatus());
    }
    /**
     * 顺序发送消息
     */
    @GetMapping("/provide/sendFor")
    public void sendFor(){
        for (int i = 0; i < 10; i++) {
//            rocketMQTemplate.syncSend("powernodes",i);
//            指定发送到broker里面的queue
            rocketMQTemplate.syncSendOrderly("powernodes","1",String.valueOf(i));
        }
    }
    /**
     * 批量发送消息
     */
    @GetMapping("/provide/sendList")
    public void sendList(){
        List<Message<String>> messageList = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Message<String> rocketMessage = MessageBuilder.withPayload("批量测试" + i)
                    // 设置消息类型
                    .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
                    .build();
            // 加入到列表
            messageList.add(rocketMessage);
        }
        //同步执行
        rocketMQTemplate.syncSend("powernodes",messageList);
        //异步
        rocketMQTemplate.asyncSend("powernodes", messageList, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("成功");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("失败");
            }
        });
        //顺序
        rocketMQTemplate.syncSendOrderly("powernodes", messageList,"1");
    }
    /**
     * tag的基本使用
     */
    @GetMapping("/provide/sendTag")
    public void sendTag(){
//        在: 后面的即为tag
        rocketMQTemplate.syncSend("powernodes:testTag","测试tag");

    }
    @GetMapping("/provide/key")
    public void sendKey(){
        Message<String> message = MessageBuilder.withPayload("我是一个带key的消息")
                .setHeader(RocketMQHeaders.KEYS, "keys")
                .setHeader(RocketMQHeaders.TAGS,"testTag")
                .build();
        rocketMQTemplate.send(message);
    }

    @GetMapping("/provide/trys")
    public String sendTrys(){
//        原生方式实现
//        DefaultMQProducer producer = rocketMQTemplate.getProducer();
//        // 设置同步发送重试次数（默认：2）
//        producer.setRetryTimesWhenSendFailed(2);
//        // 设置一般发送重试次数（默认：2）
//        producer.setRetryTimesWhenSendAsyncFailed(2);
//        producer.send();

//      采用逻辑方式实现最大重试次数以及重试间隔
        int maxRetries = 3; // 设置最大重试次数
        String topic = "powernode";
        String message = "我是一个简单的消息";

        for (int i = 0; i <= maxRetries; i++) {
            try {
                SendResult sendResult = rocketMQTemplate.syncSend(topic, message);
                if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                    return "Message sent successfully with msgId: " + sendResult.getMsgId();
                }
            } catch (Exception e) {
                if (i == maxRetries) {
                    throw e; // 如果达到最大重试次数，抛出异常
                }
                try {
                    Thread.sleep(1000 * (i + 1)); // 逐渐增加重试间隔时间
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
        }
        return "Message sending failed after retries";

    }
}
